package com.spring.mao.rabbitmq;

import com.rabbitmq.client.Channel;
import com.spring.mao.config.RabbitConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.io.IOException;

/**
 * 类功能描述：<br>
 * <ul>
 * <li>类功能描述1<br>
 * <li>类功能描述2<br>
 * <li>类功能描述3<br>
 * </ul>
 * 修改记录：<br>
 * <ul>
 * <li>修改记录描述1<br>
 * <li>修改记录描述2<br>
 * <li>修改记录描述3<br>
 * </ul>
 *
 */
@Component
public class MsgConsumer {

    private final Logger log= LoggerFactory.getLogger(this.getClass());

    @Autowired
    private StringRedisTemplate redisTemplate;

    @RabbitListener(
            bindings =
                    {
                            @QueueBinding(value = @Queue(value = RabbitConfig.FANOUT_QUEUE_NAME, durable = "true"),
                                    exchange = @Exchange(value = RabbitConfig.TEST_FANOUT_EXCHANGE, type = "fanout"))
                    })
    @RabbitHandler
    public void processFanoutMsg(Message massage, Channel channel) throws IOException {
        Long tag = massage.getMessageProperties().getDeliveryTag();
        String msgId = massage.getMessageProperties().getMessageId();

        if (redisTemplate.opsForHash().entries("fanout_queue").containsKey(msgId)) {
            log.info(msgId + ":消息已经被消费");
            channel.basicAck(tag,false);
            return;
        }
        try {
            String msg = massage.getBody().toString();
            log.info("received Fanout message : " + msg);
            redisTemplate.opsForHash().put("fanout_queue",msgId,"msgId");
            // 采用手动应答模式, 手动确认应答更为安全稳定
            channel.basicAck(tag, false);
        } catch (IOException e) {
            channel.basicNack(tag,false,true);
        }
    }

    @RabbitListener(
            bindings =
                    {
                            @QueueBinding(value = @Queue(value = RabbitConfig.FANOUT_QUEUE_NAME1, durable = "true"),
                                    exchange = @Exchange(value = RabbitConfig.TEST_FANOUT_EXCHANGE, type = "fanout"))
                    })
    @RabbitHandler
    public void processFanout1Msg(Message massage,Channel channel) throws IOException {
        Long tag = massage.getMessageProperties().getDeliveryTag();
        String msgId = massage.getMessageProperties().getMessageId();

        try {
            if (redisTemplate.opsForHash().entries("fanout_queue").containsKey(msgId)) {
                log.info(msgId + ":消息已经被消费");
                channel.basicAck(tag,false);
                return;
            }
            String msg = massage.getBody().toString();
            log.info("received Fanout1 message : " + msg);
            redisTemplate.opsForHash().put("fanout_queue",msgId,msgId);
            channel.basicAck(tag,false);
        } catch (IOException e) {
            channel.basicNack(tag,false,true);
        }

    }

    @RabbitListener(
            bindings =
                    {
                            @QueueBinding(value = @Queue(value = RabbitConfig.DIRECT_QUEUE_NAME, durable = "true"),
                                    exchange = @Exchange(value = RabbitConfig.TEST_DIRECT_EXCHANGE),
                                    key = RabbitConfig.DIRECT_ROUTINGKEY)
                    })
    @RabbitHandler
    public void processDirectMsg(Message massage,Channel channel) throws IOException {
        Long tag = massage.getMessageProperties().getDeliveryTag();
        String msgId = massage.getMessageProperties().getMessageId();


        try {
            if (redisTemplate.opsForHash().entries("direct_queue").containsKey(msgId)) {
                log.info(msgId + ":消息已经被消费");
                channel.basicAck(tag,false);
                return;
            }
            String msg =  massage.getBody().toString();
            log.info("received Direct message : " + msg);
            redisTemplate.opsForHash().put("direct_queue",msgId,msgId);

            channel.basicAck(tag,false);
        } catch (IOException e) {
            channel.basicNack(tag,false,true);
        }

    }

    @RabbitListener(
            bindings =
                    {
                            @QueueBinding(value = @Queue(value = RabbitConfig.TOPIC_QUEUE_NAME, durable = "true"),
                                    exchange = @Exchange(value = RabbitConfig.TEST_TOPIC_EXCHANGE, type = "topic"),
                                    key = RabbitConfig.TOPIC_ROUTINGKEY)
                    })
    @RabbitHandler
    public void processTopicMsg(Message massage,Channel channel) throws IOException {
        Long tag = massage.getMessageProperties().getDeliveryTag();
        String msgId = massage.getMessageProperties().getMessageId();

        try {
            if (redisTemplate.opsForHash().entries("topic_queue").containsKey(msgId)) {
                log.info(msgId + ":消息已经被消费");
                channel.basicAck(tag,false);
                return;
            }

            String msg =  massage.getBody().toString();
            log.info("received Topic message : " + msg);
            redisTemplate.opsForHash().put("topic_queue",msgId,msgId);
            channel.basicAck(tag,false);
        } catch (IOException e) {
            channel.basicNack(tag,false,true);
        }

    }

}
